package org.angelica.aliyun.manager;

import com.alicloud.openservices.tablestore.ClientConfiguration;
import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.core.utils.Pair;
import com.alicloud.openservices.tablestore.core.utils.Preconditions;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.model.filter.Filter;
import com.alicloud.openservices.tablestore.model.search.SearchQuery;
import com.alicloud.openservices.tablestore.model.search.SearchRequest;
import com.alicloud.openservices.tablestore.model.search.SearchResponse;
import com.alicloud.openservices.tablestore.model.search.query.BoolQuery;
import com.alicloud.openservices.tablestore.model.search.query.Query;
import com.aliyun.opensearch.sdk.dependencies.com.google.common.collect.Lists;
import org.angelica.aliyun.entity.StsToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 表格存储服务
 * @author aizhimin
 */

public class TableStoreManager {
    private static final Logger logger = LoggerFactory.getLogger(TableStoreManager.class);

    private static TableStoreManager instance = new TableStoreManager();
    private TableStoreManager(){ }

    private static StsToken tbStsToken = null;
    private static SyncClient syncClient = null;
    public static TableStoreManager getInstance(String endpoint, String instanceName, StsToken stsToken){
        //token过期，则重新创建SyncClient
        if(tbStsToken != null && !tbStsToken.getSecurityToken().equals(stsToken.getSecurityToken())){
            if(syncClient != null){
                syncClient.shutdown();
            }
            syncClient = null;
        }

        if(syncClient == null) {
            tbStsToken = stsToken;
            // ClientConfiguration提供了很多配置项，以下只列举部分。
            ClientConfiguration clientConfiguration = new ClientConfiguration();
            // 设置建立连接的超时时间。
            clientConfiguration.setConnectionTimeoutInMillisecond(10000);
            // 设置socket超时时间。
            clientConfiguration.setSocketTimeoutInMillisecond(10000);
            // 设置重试策略，若不设置，采用默认的重试策略。
            clientConfiguration.setRetryStrategy(new AlwaysRetryStrategy());
            syncClient = new SyncClient(endpoint, stsToken.getAccessKeyId(), stsToken.getAccessKeySecret(),
                    instanceName,clientConfiguration,stsToken.getSecurityToken());
        }
        return instance;
    }

    public static SyncClient getSyncClient(){
        return syncClient;
    }

    public Row getRow(String tableName, PrimaryKey primaryKey, Filter filter, String ...columns) {
        // 读一行
        SingleRowQueryCriteria criteria = new SingleRowQueryCriteria(tableName, primaryKey);
        // 设置读取最新版本
        criteria.setMaxVersions(1);
        // 设置过滤器
        if(filter != null) {
            criteria.setFilter(filter);
        }
        // 设置读取某些列
        if(columns != null && columns.length > 0) {
            for(String column:columns) {
                criteria.addColumnsToGet(column);
            }
        }
        GetRowResponse getRowResponse = syncClient.getRow(new GetRowRequest(criteria));
        return getRowResponse.getRow();
    }

    public List<Row> batchGetRow(String tableName, List<PrimaryKey> primaryKeyList, Filter filter, String... columns) {
        MultiRowQueryCriteria multiRowQueryCriteria = new MultiRowQueryCriteria(tableName);
        for(PrimaryKey pk : primaryKeyList) {
            multiRowQueryCriteria.addRow(pk);
        }
        // 设置读取最新版本
        multiRowQueryCriteria.setMaxVersions(1);
        // 设置过滤器
        if(filter != null) {
            multiRowQueryCriteria.setFilter(filter);
        }
        // 设置读取某些列
        if(columns != null && columns.length > 0) {
            multiRowQueryCriteria.addColumnsToGet(columns);
        }
        BatchGetRowRequest batchGetRowRequest = new BatchGetRowRequest();
        batchGetRowRequest.addMultiRowQueryCriteria(multiRowQueryCriteria);
        BatchGetRowResponse batchGetRowResponse = syncClient.batchGetRow(batchGetRowRequest);

        List<Row> rowList = Lists.newArrayList();
        if (batchGetRowResponse.isAllSucceed()) {
            for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getSucceedRows()) {
                rowList.add(rowResult.getRow());
            }
        }else {
            for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getFailedRows()) {
                System.out.println("失败的行:" + batchGetRowRequest.getPrimaryKey(rowResult.getTableName(), rowResult.getIndex()));
                System.out.println("失败原因:" + rowResult.getError());
            }
        }
        return rowList;
    }

    public List<Row> getRange(String tableName, PrimaryKey startPrimaryKey, PrimaryKey endPrimaryKey, Direction direction, Filter filter,
                              String... columns) {
        RangeRowQueryCriteria rangeRowQueryCriteria = new RangeRowQueryCriteria(tableName);
        // 设置起始主键
        rangeRowQueryCriteria.setInclusiveStartPrimaryKey(startPrimaryKey);
        // 设置结束主键
        rangeRowQueryCriteria.setExclusiveEndPrimaryKey(endPrimaryKey);
        // 设置读取最新版本
        rangeRowQueryCriteria.setMaxVersions(1);
        //倒序返回
        rangeRowQueryCriteria.setDirection(direction);
        // 设置过滤器
        if(filter != null) {
            rangeRowQueryCriteria.setFilter(filter);
        }
        // 设置读取某些列
        if(columns != null && columns.length > 0) {
            rangeRowQueryCriteria.addColumnsToGet(columns);
        }

        List<Row> rowList = Lists.newArrayList();
        while (true) {
            GetRangeResponse getRangeResponse = syncClient.getRange(new GetRangeRequest(rangeRowQueryCriteria));
            rowList.addAll(getRangeResponse.getRows());
            // 若nextStartPrimaryKey不为null, 则继续读取.
            if (getRangeResponse.getNextStartPrimaryKey() != null) {
                rangeRowQueryCriteria.setInclusiveStartPrimaryKey(getRangeResponse.getNextStartPrimaryKey());
            } else {
                break;
            }
        }

        return rowList;
    }

    /**
     * 范围查询指定范围内的数据，返回指定页数大小的数据，并能根据offset跳过部分行。
     */
    public Pair<List<Row>, PrimaryKey> getByPage(String tableName, PrimaryKey startKey, PrimaryKey endKey, Filter filter, int offset, int pageSize, Direction direction, String... columns) {
        Preconditions.checkArgument(offset >= 0, "Offset should not be negative.");
        Preconditions.checkArgument(pageSize > 0, "Page size should be greater than 0.");

        List<Row> rows = new ArrayList<Row>(pageSize);
        int limit = pageSize;
        int skip = offset;

        PrimaryKey nextStart = startKey;
        while (limit > 0 && nextStart != null) {
            RangeRowQueryCriteria criteria = new RangeRowQueryCriteria(tableName);
            criteria.setInclusiveStartPrimaryKey(nextStart);
            criteria.setExclusiveEndPrimaryKey(endKey);
            criteria.setLimit(skip + limit);
            //最新版本
            criteria.setMaxVersions(1);
            //倒序返回
            criteria.setDirection(direction);
            // 设置过滤器
            if(filter != null) {
                criteria.setFilter(filter);
            }
            // 设置读取某些列
            if(columns != null && columns.length > 0) {
                criteria.addColumnsToGet(columns);
            }

            GetRangeRequest request = new GetRangeRequest();
            request.setRangeRowQueryCriteria(criteria);
            GetRangeResponse response = syncClient.getRange(request);
            for (Row row : response.getRows()) {
                if (skip > 0) {
                    skip--;
                } else {
                    rows.add(row);
                    limit--;
                }
            }

            nextStart = response.getNextStartPrimaryKey();
        }

        return new Pair<List<Row>, PrimaryKey>(rows, nextStart);
    }

    /**
     * 插入一行
     * @param tableName
     * @param primaryKey
     * @param columns
     */
    public PrimaryKey putRow(String tableName, PrimaryKey primaryKey, List<Column> columns) {
        RowPutChange rowPutChange = new RowPutChange(tableName, primaryKey);
        //设置返回值
        rowPutChange.setReturnType(ReturnType.RT_PK);
        // 设置数据列
        rowPutChange.addColumns(columns);
        PutRowResponse response = syncClient.putRow(new PutRowRequest(rowPutChange));
        Row returnRow = response.getRow();
        if(returnRow != null){
            return returnRow.getPrimaryKey();
        }
        return null;
    }

    /**
     * 更新一行
     * @param tableName
     * @param primaryKey
     * @param columns
     */
    public void updateRow(String tableName, PrimaryKey primaryKey, List<Column> columns) {
        RowUpdateChange rowUpdateChange = new RowUpdateChange(tableName, primaryKey);
        // 设置更新一些列
        rowUpdateChange.put(columns);
        syncClient.updateRow(new UpdateRowRequest(rowUpdateChange));
    }

    /**
     * 条件更新
     * @param tableName
     * @param primaryKey
     * @param columns
     * @param condition
     */
    public void updateRowWithCondition(String tableName, PrimaryKey primaryKey, List<Column> columns, Condition condition){
        try {
            RowUpdateChange rowUpdateChange = new RowUpdateChange(tableName, primaryKey);
            rowUpdateChange.setCondition(condition);
            rowUpdateChange.put(columns);
            syncClient.updateRow(new UpdateRowRequest(rowUpdateChange));
        }catch (Exception e){
            logger.error(e.getMessage());
        }
    }

    /**
     * 删除一行
     * @param tableName
     * @param primaryKey
     */
    public void deleteRow(String tableName, PrimaryKey primaryKey) {
        RowDeleteChange rowDeleteChange = new RowDeleteChange(tableName, primaryKey);
        syncClient.deleteRow(new DeleteRowRequest(rowDeleteChange));
    }

    /**
     * 批量更新
     * @param tableName
     * @param rowList
     * @return
     */
    public List<PrimaryKey> batchWriteRow(String tableName, List<Map<PrimaryKey, List<Column>>> rowList) {
        BatchWriteRowRequest batchWriteRowRequest = new BatchWriteRowRequest();

        for(Map<PrimaryKey, List<Column>> map:rowList) {
            PrimaryKey pk = map.keySet().iterator().next();
            RowUpdateChange rowUpdateChange = new RowUpdateChange(tableName, pk);
            rowUpdateChange.put(map.get(pk));
            batchWriteRowRequest.addRowChange(rowUpdateChange);
        }

        BatchWriteRowResponse response = syncClient.batchWriteRow(batchWriteRowRequest);
        List<PrimaryKey> failedRows = Lists.newArrayList();
        if (!response.isAllSucceed()) {
            for (BatchWriteRowResponse.RowResult rowResult : response.getFailedRows()) {
                PrimaryKey pk = batchWriteRowRequest.getRowChange(rowResult.getTableName(), rowResult.getIndex()).getPrimaryKey();
                failedRows.add(pk);
                logger.info("pk:{},error:{}",pk,rowResult.getError());
            }
        }
        return failedRows;
    }

    /**
     * 批量删除行
     * @param tableName
     * @param primaryKeyList
     */
    public void batchDeleteRow(String tableName, List<PrimaryKey> primaryKeyList) {
        BatchWriteRowRequest batchWriteRowRequest = new BatchWriteRowRequest();

        for(PrimaryKey pk:primaryKeyList) {
            RowDeleteChange rowDeleteChange = new RowDeleteChange(tableName, pk);
            batchWriteRowRequest.addRowChange(rowDeleteChange);
        }

        BatchWriteRowResponse response = syncClient.batchWriteRow(batchWriteRowRequest);
        List<PrimaryKey> failedRows = Lists.newArrayList();
        if (!response.isAllSucceed()) {
            for (BatchWriteRowResponse.RowResult rowResult : response.getFailedRows()) {
                PrimaryKey pk = batchWriteRowRequest.getRowChange(rowResult.getTableName(), rowResult.getIndex()).getPrimaryKey();
                failedRows.add(pk);
                logger.info("pk:{},error:{}",pk,rowResult.getError());
            }
        }
    }

    public void createTable(TableMeta tableMeta) {

        // 永不过期，也可以设置数据有效期，过期了会自动删除
        int timeToLive = -1;
        // 只保存一个版本，目前支持多版本
        int maxVersions = 1;

        TableOptions tableOptions = new TableOptions(timeToLive, maxVersions);

        CreateTableRequest request = new CreateTableRequest(tableMeta, tableOptions);

        syncClient.createTable(request);
    }

    /**
     * 多条件查询
     * @param tableName
     * @param indexName
     * @param boolQuery
     * @param columns
     * @return
     */
    public SearchResponse boolQuery(String tableName,String indexName,BoolQuery boolQuery,List<String> columns){
        SearchQuery searchQuery = new SearchQuery();
        searchQuery.setQuery(boolQuery);
        searchQuery.setGetTotalCount(true);
        SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
        SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
        if(columns != null && !columns.isEmpty()){
            //设置返回的列
            columnsToGet.setColumns(columns);
        }else{
            // 设置返回所有列
            columnsToGet.setReturnAll(true);
        }
        searchRequest.setColumnsToGet(columnsToGet);
        return syncClient.search(searchRequest);
    }

    /**
     * 查询数据总行数
     * @param tableName
     * @param indexName
     * @return
     */
    public long getTotalCount(String tableName, String indexName, Query query){
        SearchQuery searchQuery = new SearchQuery();
        searchQuery.setQuery(query);
        searchQuery.setLimit(0);
        searchQuery.setGetTotalCount(true);
        SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
        SearchResponse searchResponse = syncClient.search(searchRequest);
        return searchResponse.getTotalCount();
    }

}
